package com.learning.hadoop.sqoop.thread;

import com.google.common.collect.Lists;

import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * ClassName: Worker
 * Description: 工作线程
 * Date: 2018/4/18 10:53 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class Worker implements Runnable {
    private ArrayBlockingQueue<Object> queue;
    private int batchSize;


    public Worker(ArrayBlockingQueue<Object> queue, int batchSize) {
        this.queue = queue;
        this.batchSize = batchSize;
    }

    @Override
    public void run() {
        long start = System.currentTimeMillis();

        while (true) {
//            if (!queue.isEmpty()) {
            // 存放最终的结果
            ArrayList<Object> resultList = Lists.newArrayListWithExpectedSize(batchSize);
            Object poll = queue.poll();
//                if (poll != null) {
            try {
                resultList.add(Cal.cal(poll));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
//                }

            long end = System.currentTimeMillis();
            if (resultList.size() == batchSize || (end - start) > 1000 * 2L) {
                // TODO 保存数据库
                System.out.println("!~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
                Cal.save(resultList);
                start = System.currentTimeMillis();
            }
        }
//        }


    }
}
